home *** CD-ROM | disk | FTP | other *** search
/ Sprite 1984 - 1993 / Sprite 1984 - 1993.iso / src / kernel / rpc / rpcClient.c < prev    next >
Encoding:
C/C++ Source or Header  |  1992-12-19  |  20.1 KB  |  649 lines
  1. /*
  2.  * rpcClient.c --
  3.  *
  4.  *      The client side of the RPC protocol.  The routines here are in
  5.  *      this file because they synchronize with each other using a master
  6.  *      lock.  RpcDoCall is the send-receive-timeout loop and
  7.  *      RpcClientDispatch is the interrupt time routine that gets packets
  8.  *      and hands them up to RpcDoCall.
  9.  *
  10.  * Copyright 1986 Regents of the University of California
  11.  * All rights reserved.
  12.  */
  13.  
  14. #ifndef lint
  15. static char rcsid[] = "$Header: /cdrom/src/kernel/Cvsroot/kernel/rpc/rpcClient.c,v 9.12 92/12/13 18:21:30 mgbaker Exp $ SPRITE (Berkeley)";
  16. #endif /* not lint */
  17.  
  18.  
  19. #include <sprite.h>
  20. #include <stdio.h>
  21. #include <rpc.h>
  22. #include <rpcInt.h>
  23. #include <rpcClient.h>
  24. #include <rpcServer.h>
  25. #include <rpcTrace.h>
  26. #include <dbg.h>
  27. #include <proc.h>
  28. #include <sys.h>
  29.  
  30. #include <recov.h>
  31.  
  32.  
  33. /*
  34.  * For debugging servers.  We allow client's to retry forever instead
  35.  * of timing out.  This is exported and settable via Fs_Command
  36.  */
  37. Boolean rpc_NoTimeouts = FALSE;
  38.  
  39. /*
  40.  * A histogram is kept of the elapsed time of each different kind of RPC.
  41.  */
  42. Rpc_Histogram *rpcCallTime[RPC_LAST_COMMAND+1];
  43. Boolean rpcCallTiming = FALSE;
  44.  
  45. #ifdef DEBUG
  46. #define DEBUGSIZE 1000
  47. #define INC(ctr) { (ctr) = ((ctr) == DEBUGSIZE-1) ? 0 : (ctr)+1; }
  48. typedef struct {
  49.     RpcClientChannel    *chanPtr;
  50.     char        *action;
  51.     int            pNum;
  52.     int            serverID;
  53.     int            chanNum;
  54.     int            state;
  55. } dbgElem;
  56.  
  57. static dbgElem    dbgArray[DEBUGSIZE];
  58. static int         dbgCtr;
  59.  
  60. #define CHAN_TRACE(zchanPtr, serverID, string) \
  61. { \
  62.     dbgElem *ptr = &dbgArray[dbgCtr]; \
  63.     INC(dbgCtr); \
  64.     ptr->chanPtr = zchanPtr; \
  65.     ptr->action = string; \
  66.     ptr->chanNum = zchanPtr->index; \
  67.     ptr->serverID = serverID; \
  68.     ptr->pNum = Mach_GetProcessorNumber(); \
  69.     ptr->state = zchanPtr->state; \
  70. }
  71.  
  72. #else
  73. #define CHAN_TRACE(zchanPtr, serverID, string)
  74. #endif
  75.  
  76. /* Variables to control nack back-off on client. */
  77. int    rpcNackRetryWait;
  78. int    rpcMaxNackWait;
  79. /*
  80.  * This variable determines whether we use client policy of ramping down
  81.  * channels for neg acks.  The default is to use backoff.
  82.  */
  83. Boolean    rpcChannelNegAcks = FALSE;
  84.  
  85.  
  86. /*
  87.  *----------------------------------------------------------------------
  88.  *
  89.  * RpcDoCall --
  90.  *
  91.  *    The send-receive-timeout loop on the client for a remote procedure call.
  92.  *
  93.  * Results:
  94.  *    The return code from the remote procedure or an error code
  95.  *    related to the RPC protocol, or SUCCESS.
  96.  *
  97.  * Side effects:
  98.  *    The remote procedure call itself.
  99.  *
  100.  *----------------------------------------------------------------------
  101.  */
  102. ReturnStatus
  103. RpcDoCall(serverID, chanPtr, storagePtr, command, srvBootIDPtr, notActivePtr,
  104.     recovTypePtr)
  105.     int serverID;        /* The Sprite host that will execute the
  106.                  * service procedure */
  107.     register RpcClientChannel *chanPtr;    /* The channel for the RPC */
  108.     Rpc_Storage *storagePtr;    /* Pointers to caller's buffers */
  109.     int command;        /* Only used to filter trace records */
  110.     unsigned    int *srvBootIDPtr;    /* Return, boot time stamp of server. */
  111.     int *notActivePtr;        /* Return, RPC_NOT_ACTIVE flag from server.
  112.                  * These last two return parameters are later
  113.                  * passed to the recovery module. */
  114.     unsigned int *recovTypePtr;    /* Wether rpc packet comes from fast booted
  115.                  * server or server doing special recovery. */
  116. {
  117.     register RpcHdr *rpcHdrPtr;    /* Pointer to received message header */
  118.     register RpcConst *constPtr;/* Timeout parameter block */
  119.     register ReturnStatus error;/* General error return status */
  120.     register unsigned int wait;    /* Wait interval for timeouts */
  121.     int numAcks;        /* Count of acks received.  Used to catch the
  122.                  * case where the server process hangs and
  123.                  * the server dispatcher acks us forever */
  124.     register int numTries;    /* Number of times we sent the message while
  125.                  * getting no reply. */
  126.     register unsigned int lastFragMask = 0;    /* Previous state of our
  127.                          * fragment reassembly */
  128.     Boolean    seemsHung = FALSE;    /* Used to control warning msgs */
  129.  
  130.     /*
  131.      * This code is locked with MASTER_LOCK in order to synchronize
  132.      * with the RpcClientDispatch routine.  Inside the critical
  133.      * section we sleep on the channel's wait condition to which
  134.      * RpcClientDispatch broadcasts when it gets input.  Furthermore,
  135.      * we place a call back procedure in the timer queue that will
  136.      * also notify that condition upon timeout.
  137.      */
  138.     MASTER_LOCK(&chanPtr->mutex);
  139.  
  140.     /*
  141.      * Send the request off to the server.  We update the server hint from
  142.      * the channel's return message header.  ie. take the last server
  143.      * hint received from the server.
  144.      */
  145.  
  146.     *srvBootIDPtr = 0;
  147.     rpcCltStat.requests++;
  148.     chanPtr->requestRpcHdr.serverHint =    chanPtr->replyRpcHdr.serverHint;
  149.     chanPtr->state |= CHAN_WAITING;
  150.     error = RpcOutput(serverID, (RpcHdr *) &chanPtr->requestRpcHdr,
  151.               &chanPtr->request, chanPtr->fragment,
  152.               (unsigned int) (chanPtr->fragsDelivered),
  153.               &chanPtr->mutex);
  154.     /*
  155.      * Set up the initial wait interval.  The wait could depend on
  156.      * characteristics of the RPC, or of the other host.
  157.      * For now we just wait longer if the packet will be fragmented.
  158.      */
  159.     constPtr = chanPtr->constPtr;
  160.     if ((storagePtr->requestDataSize + storagePtr->requestParamSize >
  161.         RPC_MAX_FRAG_SIZE) ||
  162.     (storagePtr->replyDataSize + storagePtr->replyParamSize >
  163.         RPC_MAX_FRAG_SIZE)) {
  164.     wait = constPtr->fragRetryWait;
  165.     } else {
  166.     wait = constPtr->retryWait;
  167.     }
  168.  
  169.     /*
  170.      * Loop waiting for input and re-sending if need be.  As well as
  171.      * getting a reply from the server we screen out junk and react to
  172.      * explicit acks.  This times out after a number of times around the
  173.      * loop with no input.
  174.      */
  175.     numTries = 0;
  176.     numAcks = 0;
  177.     do {
  178.  
  179.     /*
  180.      * Wait until we get a poke from the timeout routine or there
  181.      * is input available.  Input may have arrived before we get
  182.      * here because the channel mutex is released while Rpc_Output
  183.      * waits for the packet to be sent by the network interface.
  184.      */
  185.     if (! (chanPtr->state & CHAN_INPUT)) {
  186.         chanPtr->timeoutItem.routine = Rpc_Timeout;
  187.         chanPtr->timeoutItem.interval = wait;
  188.         chanPtr->timeoutItem.clientData = (ClientData)chanPtr;
  189.         chanPtr->state |= CHAN_TIMEOUT | CHAN_WAITING;
  190.         CHAN_TRACE(chanPtr, serverID, "about to schedule");
  191.         Timer_ScheduleRoutine(&chanPtr->timeoutItem, TRUE);
  192.         do {
  193.         /*
  194.          * Wait ignoring signals.
  195.          */
  196.         Sync_MasterWait(&chanPtr->waitCondition,
  197.                 &chanPtr->mutex, FALSE);
  198.         } while (((chanPtr->state & CHAN_INPUT) == 0) &&
  199.              (chanPtr->state & CHAN_TIMEOUT));
  200.         CHAN_TRACE(chanPtr, serverID, "woken up");
  201.     }
  202.     if (chanPtr->state & CHAN_INPUT) {
  203.         /*
  204.          * Got some input.  The dispatch routine has copied the
  205.          * packet into the areas refered to by the reply BufferSet.
  206.          *
  207.          * NB: We have to completely process this message before we
  208.          * can accept another message on this channel.  There is no
  209.          * mechanism to queue messages.
  210.          */
  211.         chanPtr->state &= ~CHAN_INPUT;
  212.         rpcHdrPtr = &chanPtr->replyRpcHdr;
  213.         *recovTypePtr = 0;
  214.         if (rpcHdrPtr->flags & RPC_FAST) {
  215.         *recovTypePtr |= RECOV_FAST_BOOT;
  216.         }
  217.         if (rpcHdrPtr->flags & RPC_SERVER_RECOV) {
  218.         *recovTypePtr |= RECOV_SERVER_DRIVEN;
  219.         }
  220.         /*
  221.          * Pick off the boot timestamp and active state of the server so
  222.          * the recovery module can pay attention to traffic.
  223.          */
  224.         *notActivePtr = rpcHdrPtr->flags & RPC_NOT_ACTIVE;
  225.         *srvBootIDPtr = rpcHdrPtr->bootID;
  226. #ifdef TIMESTAMP
  227.         RPC_TRACE(rpcHdrPtr, RPC_CLIENT_D, "input");
  228. #endif /* TIMESTAMP */
  229.         if (rpcHdrPtr->ID != chanPtr->requestRpcHdr.ID) {
  230.         /*
  231.          * Note old message.
  232.          */
  233.         rpcCltStat.oldInputs++;
  234.  
  235.         } else if (rpcHdrPtr->flags & RPC_NACK) {
  236.         rpcCltStat.nacks++;
  237.         /*
  238.          * Try out different nack-handling policies.  
  239.          * We can either back off as in an ACK, or try to ramp
  240.          * down the number of channels.
  241.          */
  242.         if (serverID != RPC_BROADCAST_SERVER_ID) {
  243.             numTries = 0;
  244.             if (!rpcChannelNegAcks) {
  245.             if (wait < rpcNackRetryWait) {
  246.                 wait = rpcNackRetryWait;
  247.             } else {
  248.                 Net_HostPrint(serverID,
  249.                 "Client backing off again from negative ack.\n");
  250.                 wait *= 2;
  251.                 rpcCltStat.reNacks++;
  252.             }
  253.             if (wait > rpcMaxNackWait) {
  254.                 Net_HostPrint(serverID,
  255.                 "Client setting max backoff from negative ack.\n");
  256.                 wait = rpcMaxNackWait;
  257.                 rpcCltStat.maxNacks++;
  258.             }
  259.             } else {
  260.             /* Return error to cause ramping down of channels. */
  261.             error = RPC_NACK_ERROR; 
  262.             }
  263.         }
  264.         } else if (rpcHdrPtr->flags & RPC_REPLY) {
  265.         /*
  266.          * Our reply, check for an error code and break from the
  267.          * receive loop.  The command field is overloaded with the
  268.          * return error code.
  269.          */
  270.         if (rpcHdrPtr->flags & RPC_ERROR) {
  271.             error = (ReturnStatus)rpcHdrPtr->command;
  272.             if (error == 0) {
  273.             rpcCltStat.nullErrors++;
  274.             error = RPC_NULL_ERROR;
  275.             } else {
  276.             rpcCltStat.errors++;
  277.             }
  278.         } else {
  279.             rpcCltStat.replies++;
  280.         }
  281.         /*
  282.          * Copy back the return buffer size (it's set by ClientDispatch)
  283.          * to reflect what really came back.
  284.          */
  285.         storagePtr->replyDataSize = chanPtr->actualDataSize;
  286.         storagePtr->replyParamSize = chanPtr->actualParamSize;
  287.         break;
  288.         } else if (rpcHdrPtr->flags & RPC_ACK) {
  289.         numAcks++;
  290.         rpcCltStat.acks++;
  291.         if (numAcks <= constPtr->maxAcks) {
  292.             /*
  293.              * An ack from the server indicating that a server
  294.              * process is working on our request.  We increase
  295.              * our waiting time to decrease the change that we'll
  296.              * timeout again before receiving the reply.
  297.              * NOTE: We don't pay attention to acks if we are
  298.              * broadcasting.  This makes the broadcaster too vulnerable
  299.              * to errant servers.  In particular, diskless clients
  300.              * often wedge trying to handle a prefix request, send
  301.              * the real server a lot of acks, and slow it's boot down.
  302.              */
  303.             if (serverID != RPC_BROADCAST_SERVER_ID) {
  304.             numTries = 0;
  305.             wait *= 2;
  306.             if (wait > constPtr->maxAckWait) {
  307.                 wait = constPtr->maxAckWait;
  308.             }
  309.             }
  310.         } else {
  311.             char name[100];
  312.             /*
  313.              * Too many acks.  It is very likely that the server
  314.              * process is hung on some lock.  We hang too in
  315.              * order to facilitate debugging.
  316.              */
  317.             rpcCltStat.tooManyAcks++;
  318.             if (!seemsHung) {
  319.             Net_SpriteIDToName(serverID, 100, name);
  320.             if (name == (char *)NIL) {
  321.                 printf("RpcDoCall: <%s> RPC to host <%d> is hung\n",
  322.                 rpcService[command].name, serverID);
  323.             } else {
  324.                 printf("RpcDoCall: <%s> RPC to %s is hung\n",
  325.                 rpcService[command].name, name);
  326.             }
  327.             seemsHung = TRUE;
  328.             }
  329.             numAcks = 0;
  330.         }
  331.         } else {
  332.         /*
  333.          * Unexpected kind of input
  334.          */
  335.         rpcCltStat.badInput++;
  336.         printf("Warning: Rpc_Call: Unexpected input.\n");
  337.         error = RPC_INTERNAL_ERROR;
  338.         }
  339.     } else {
  340.         /*
  341.          * Have not received a complete message yet.  Update the
  342.          * server hint and then re-send the request or send a partial
  343.          * acknowledgment.
  344.          */
  345.         chanPtr->requestRpcHdr.serverHint =
  346.         chanPtr->replyRpcHdr.serverHint;
  347.         rpcCltStat.timeouts++;
  348.         /*
  349.          * Back off upon timeout because we may be talking to a slow host
  350.          */
  351.         wait *= 2;
  352.         if (wait > constPtr->maxTimeoutWait) {
  353.         wait = constPtr->maxTimeoutWait;
  354.         }
  355.         if ((chanPtr->state & CHAN_FRAGMENTING) == 0) {
  356.         /*
  357.          * Not receiving fragments.  Check for timeout, and resend
  358.          * the request.
  359.          */
  360.         numTries++;
  361.         if (numTries < constPtr->maxTries ||
  362.             (rpc_NoTimeouts && serverID != RPC_BROADCAST_SERVER_ID)) {
  363.             rpcCltStat.resends++;
  364.             chanPtr->requestRpcHdr.flags |= RPC_PLSACK;
  365.             error = RpcOutput(serverID, 
  366.                       (RpcHdr *) &chanPtr->requestRpcHdr,
  367.                       &chanPtr->request, chanPtr->fragment,
  368.                       (unsigned int) (chanPtr->fragsDelivered),
  369.                       &chanPtr->mutex);
  370.         } else {
  371.             rpcCltStat.aborts++;
  372.             error = RPC_TIMEOUT;
  373.         }
  374.         } else {
  375.         /*
  376.          * We are getting a fragmented response.  The client dispatcher
  377.          * has set the fragsReceived field to reflect the state
  378.          * of fragment reassembly.  We check that and will abort
  379.          * if we timeout too many times with no new fragments.
  380.          * Otherwise we return a partial acknowledgment.
  381.          */
  382.         if (lastFragMask < chanPtr->fragsReceived) {
  383.             lastFragMask = chanPtr->fragsReceived;
  384.             numTries = 0;
  385.         } else {
  386.             numTries++;
  387.         }
  388.         if (numTries >= constPtr->maxTries &&
  389.             (!rpc_NoTimeouts||(serverID == RPC_BROADCAST_SERVER_ID))) {
  390.             rpcCltStat.aborts++;
  391.             error = RPC_TIMEOUT;
  392.         } else {
  393.             chanPtr->requestRpcHdr.flags = RPC_SERVER |RPC_ACK |
  394.                             RPC_LASTFRAG;
  395.             chanPtr->requestRpcHdr.fragMask = chanPtr->fragsReceived;
  396.             rpcCltStat.sentPartial++;
  397.             error = RpcOutput(serverID, 
  398.                       (RpcHdr *) &chanPtr->requestRpcHdr,
  399.                       &chanPtr->request, chanPtr->fragment,
  400.                       (unsigned int) (chanPtr->fragsDelivered),
  401.                       &chanPtr->mutex);
  402.         }
  403.         }
  404.     }
  405.     } while (error == SUCCESS);
  406.     chanPtr->state &= ~CHAN_WAITING;
  407.     if (seemsHung) {
  408.     if (error == SUCCESS) {
  409.         printf("<%s> RPC ok\n", rpcService[command].name);
  410.     } else {
  411.         printf("<%s> RPC exit 0x%x\n", rpcService[command].name, error);
  412.     }
  413.     }
  414.     MASTER_UNLOCK(&chanPtr->mutex);
  415.     return(error);
  416. }
  417.  
  418. /*
  419.  *----------------------------------------------------------------------
  420.  *
  421.  * RpcClientDispatch --
  422.  *
  423.  *      Dispatch a message to a client channel.  The channel has buffer
  424.  *      space for the packet header and also specifies the buffer areas
  425.  *      for the RPC return parameters and data.  We notify the owner of
  426.  *      the channel that is has input via the condition variable in the
  427.  *      channel.
  428.  *
  429.  * Sprite Id:
  430.  *    This routine has the side effect of initializing the Sprite ID
  431.  *    of the host from information in the RPC packet header.  This is
  432.  *    for diskless clients of the filesystem that have no other way
  433.  *    to determine their Sprite ID.  The routine RpcValidateClient
  434.  *    on the server side initializes the clientID field for us.
  435.  *
  436.  * Results:
  437.  *    None.
  438.  *
  439.  * Side effects:
  440.  *    The message is copied into the buffers specified by the channel
  441.  *    and the channel is notified of input.
  442.  *
  443.  *----------------------------------------------------------------------
  444.  */
  445. void
  446. RpcClientDispatch(chanPtr, rpcHdrPtr)
  447.     register RpcClientChannel *chanPtr;    /* The channel the packet is for */
  448.     register RpcHdr *rpcHdrPtr;        /* The Rpc header as it sits in the
  449.                      * network module's buffer.  The data
  450.                      * in the message follows this. */
  451. {
  452.     register int size;    /* The amount of data in the message */
  453.     /*
  454.      * Acquire the channel mutex for multiprocessor synchronization.
  455.      * Deadlock occurs here when doing RPCs to oneself and the client
  456.      * resends and then the server acknowledges.
  457.      */
  458. #if (MACH_MAX_NUM_PROCESSORS == 1) /* uniprocessor implementation */
  459.     if (chanPtr->mutex.value != 0) {
  460.     printf("Warning:  Rpc to myself?\n");
  461.     return;
  462.     } else {
  463.     MASTER_LOCK(&chanPtr->mutex);
  464.     }
  465. #else    /* Multiprocessor implementation. */
  466.     MASTER_LOCK(&chanPtr->mutex);
  467. #endif
  468.     
  469.     /*
  470.      * Discover our own Sprite ID from the packet header.
  471.      */
  472.     if (rpc_SpriteID == 0) {
  473.     rpc_SpriteID = rpcHdrPtr->clientID;
  474.     printf("RPC setting SpriteID to %d.\n", rpc_SpriteID);
  475.     } else if (rpc_SpriteID != rpcHdrPtr->clientID) {
  476.     printf("RpcClientDispatch: clientID changed from (%d) to (%d).\n",
  477.                        rpc_SpriteID, rpcHdrPtr->clientID);
  478.     }
  479.  
  480.     /*
  481.      * See if this is a close request from the server - the server host is
  482.      * trying to recycle the server process that is bound to this
  483.      * channel.  If the channel is not busy now it implies we have completed
  484.      * the RPC the server is inquiring about.  We return an ack to the
  485.      * server and unbind the server process from us.  During this the
  486.      * channel is marked busy so that while we are returning the ack,
  487.      * another process doesn't come along, allocate the channel, and try
  488.      * to use the same buffers as us.
  489.      */
  490.     if (rpcHdrPtr->flags & RPC_CLOSE) {
  491.     RpcChanClose(chanPtr, rpcHdrPtr);
  492.     goto unlock;
  493.     }
  494.  
  495.     /*
  496.      * Verify the transaction Id.  Doing this now after checking for
  497.      * close messages means we still ack close requests even if
  498.      * we have already recycled the channel.
  499.      */
  500.     if (rpcHdrPtr->ID != chanPtr->requestRpcHdr.ID) {
  501.     rpcCltStat.badId++;
  502.     goto unlock;
  503.     }
  504.  
  505.     /*
  506.      * Filter out partial acks.
  507.      */
  508.     if ((rpcHdrPtr->fragMask != 0) && (rpcHdrPtr->flags & RPC_ACK)) {
  509.     if (chanPtr->fragsDelivered != rpcHdrPtr->fragMask) {
  510.         /*
  511.          * Because we may get several partial acks, we just update
  512.          * fragsDelivered so the next resend is smarter.  Eventually
  513.          * we'll get all the fragments delivered and then these
  514.          * acks will get passed up to RpcDoCall().
  515.          */
  516.         chanPtr->fragsDelivered = rpcHdrPtr->fragMask;
  517.         rpcCltStat.recvPartial++;
  518.         goto unlock;
  519.     }
  520.     /*
  521.      * Apparently we've gotten everything through to the
  522.      * server.  Our last transmission was a keep-alive
  523.      * of just the last fragment.  We FALL THROUGH to
  524.      * pass the server's ack up to the process level, RpcDoCall();
  525.      */
  526.     }
  527.  
  528.     /*
  529.      * See if the channel is available for input.  Currently there
  530.      * is no queueing of packets so we drop this packet if the process
  531.      * is still working on the last packet it got.
  532.      */
  533.     if ((chanPtr->state & CHAN_WAITING) == 0) {
  534.     rpcCltStat.chanBusy++;
  535.     goto unlock;
  536.     }
  537.     /*
  538.      * Note for RpcDoCall the actual size of the returned information.
  539.      */
  540.     size = rpcHdrPtr->paramSize + rpcHdrPtr->paramOffset;
  541.     if (chanPtr->actualParamSize < size) {
  542.     chanPtr->actualParamSize = size;
  543.     }
  544.     size = rpcHdrPtr->dataSize + rpcHdrPtr->dataOffset;
  545.     if (chanPtr->actualDataSize < size) {
  546.     chanPtr->actualDataSize = size;
  547.     }
  548.  
  549.     if (rpcHdrPtr->numFrags != 0) {
  550.     if ((chanPtr->state & CHAN_FRAGMENTING) == 0) {
  551.         /*
  552.          * The first fragment of a fragmented reply.
  553.          */
  554.         RpcScatter(rpcHdrPtr, &chanPtr->reply);
  555.         chanPtr->fragsReceived = rpcHdrPtr->fragMask;
  556.         chanPtr->state |= CHAN_FRAGMENTING;
  557.         goto unlock;
  558.     } else if (chanPtr->fragsReceived & rpcHdrPtr->fragMask) {
  559.         /*
  560.          * Duplicate fragment.
  561.          */
  562.         rpcCltStat.dupFrag++;
  563.         goto unlock;
  564.     } else {
  565.         /*
  566.          * More fragments.
  567.          */
  568.         RpcScatter(rpcHdrPtr, &chanPtr->reply);
  569.         chanPtr->fragsReceived |= rpcHdrPtr->fragMask;
  570.         if (chanPtr->fragsReceived !=
  571.            rpcCompleteMask[rpcHdrPtr->numFrags]) {
  572.         goto unlock;
  573.         } else {
  574.         /*
  575.          * Now the packet is complete.
  576.          */
  577.         chanPtr->state &= ~CHAN_FRAGMENTING;
  578.         }
  579.     }
  580.     } else {
  581.     /*
  582.      * Unfragmented message.
  583.      * Copy the complete message out of the network's buffers.
  584.      */
  585.     RpcScatter(rpcHdrPtr, &chanPtr->reply);
  586.     }
  587.     
  588.     /*
  589.      * Remove the channel from the timeout queue and
  590.      * notify the waiting channel that it has input.
  591.      */
  592.     chanPtr->state &= ~CHAN_WAITING;
  593.     chanPtr->state |= CHAN_INPUT;
  594.  
  595.     if (chanPtr->state & CHAN_TIMEOUT) {
  596.     chanPtr->state &= ~CHAN_TIMEOUT;
  597.     CHAN_TRACE(chanPtr, chanPtr->serverID, "about to deschedule");
  598.     (void)Timer_DescheduleRoutine(&chanPtr->timeoutItem);
  599.     }
  600.     Sync_MasterBroadcast(&chanPtr->waitCondition);
  601.  
  602. unlock:
  603. #ifdef TIMESTAMP
  604.     RPC_TRACE(rpcHdrPtr, RPC_CLIENT_a, "client");
  605. #endif /* TIMESTAMP */
  606.  
  607.     MASTER_UNLOCK(&chanPtr->mutex);
  608. }
  609.  
  610. /*
  611.  *----------------------------------------------------------------------
  612.  *
  613.  * Rpc_Timeout --
  614.  *
  615.  *    Called when a channel times out.  This notifies the waiting condition
  616.  *    of the channel so the process can wake up and take action upon
  617.  *    the timeout.  The mutex on the channel is aquired for multiprocessor
  618.  *    synchronization.
  619.  *
  620.  * Results:
  621.  *    None.
  622.  *
  623.  * Side effects:
  624.  *    The channel contition is notified.
  625.  *
  626.  *----------------------------------------------------------------------
  627.  */
  628. /*ARGSUSED*/
  629. void
  630. Rpc_Timeout(time, data)
  631.     Timer_Ticks time;    /* The time we timed out at. */
  632.     ClientData data;    /* Our private data is the channel pointer */
  633. {
  634.     RpcClientChannel *chanPtr;    /* The channel to notify */
  635.  
  636.     chanPtr = (RpcClientChannel *)data;
  637.  
  638.     /*
  639.      * This is called from the timeout queue at interrupt time.
  640.      * We acquire the master lock (as a formality in a uniprocessor)
  641.      * and use the form of broadcast designed for master locks.
  642.      */
  643.     MASTER_LOCK(&chanPtr->mutex);
  644.     chanPtr->state &= ~CHAN_TIMEOUT;
  645.     CHAN_TRACE(chanPtr, chanPtr->serverID, "Timeout");
  646.     Sync_MasterBroadcast(&chanPtr->waitCondition);
  647.     MASTER_UNLOCK(&chanPtr->mutex);
  648. }
  649.